package api

import bean.SensorReading
import org.apache.flink.api.common.functions.{FilterFunction, RichMapFunction}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{ConnectedStreams, DataStream, SplitStream, StreamExecutionEnvironment}

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  *
  *          多流转换操作
  */
object Transform3 {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputPath: String = "src/main/resources/sensor.txt"
    val inputStream: DataStream[String] = env.readTextFile(inputPath)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = inputStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    //分流
    val splitStream: SplitStream[SensorReading] = dataStream
      .split(data => {
        if (data.temperature > 30.0)
          Seq("high")
        else
          Seq("low")
      })
    val highTempStream: DataStream[SensorReading] = splitStream.select("high")

    val lowTempStream: DataStream[SensorReading] = splitStream.select("low")

    val allTempStream: DataStream[SensorReading] = splitStream.select("high", "low")


    //合流
    val warningStream: DataStream[(String, Double)] = highTempStream
      .map(data => (data.id, data.temperature))

    val connectedStreams: ConnectedStreams[(String, Double), SensorReading] = warningStream
      .connect(lowTempStream)

    //CoMap对数据进行分别处理
    val coMapResultStream: DataStream[Product] = connectedStreams.map(
      warningData => (warningData._1, warningData._2, "warning"),
      lowTempData => (lowTempData.id, "healthy")
    )

    //union合流(同类型)
    val unionStream: DataStream[SensorReading] = highTempStream.union(lowTempStream)

    //    highTempStream.print("high")
    //    lowTempStream.print("low")
    //    allTempStream.print("all")
    //    coMapResultStream.print("coMap")
    unionStream.print("union")

    env.execute("temperature test")

  }

  //自定义函数类
  class MyFilter extends FilterFunction[SensorReading]{
    override def filter(t: SensorReading): Boolean =
      t.id.startsWith("sensor_1")
  }

  //自定义富函数,可以获取程序运行时上下文，还有一些生命周期
  class MyRichMapper extends RichMapFunction[SensorReading,String]{

    override def open(parameters: Configuration): Unit = super.open(parameters)

    override def map(in: SensorReading): String = in.id+"temperature"
  }

}
